package com.atguigu.gmall.realtime.apps

import java.time.format.DateTimeFormatter
import java.time.{Instant, LocalDate, LocalDateTime, ZoneId}
import java.util

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.contants.GmallConstants
import com.atguigu.gmall.realtime.bean.{CouponAlertInfo, EventLog}
import com.atguigu.gmall.realtime.utils.{MyEsUtil, MykafkaUtil}
import io.searchbox.client.JestClient
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.Minutes
import org.apache.spark.streaming.dstream.{DStream, InputDStream}

import scala.util.control.Breaks
import scala.util.control.Breaks._

/**
 * Date:2021/3/25
 * Author:csw
 * Description:
 */
object AlertApp extends BaseApp {
  override val appName: String = "Alert"
  override val seound: Int = 10

  def main(args: Array[String]): Unit = {
    runApp {
      val ds: InputDStream[ConsumerRecord[String, String]] = MykafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT, ssc)
      ssc.sparkContext.setLogLevel("warn")
      //格式化
      val ds1: DStream[(String, EventLog)] = ds.map(record => {
        val eventLog: EventLog = JSON.parseObject(record.value(), classOf[EventLog])
        val time: LocalDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(eventLog.ts), ZoneId.of("Asia/Shanghai"))
        val formatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
        eventLog.logDate = time.format(formatter)
        eventLog.logHour = time.getHour + ""
        (eventLog.mid, eventLog)
      })
      //开窗，聚合
      val ds2: DStream[(String, Iterable[EventLog])] = ds1.window(Minutes(5)).groupByKey()

      //ds1.print()
      val ds3: DStream[CouponAlertInfo] = ds2.map {
        case (mid, logs) => {
          var uids: util.HashSet[String] = new java.util.HashSet[String]()
          var itemIds: util.HashSet[String] = new java.util.HashSet[String]()
          var events: util.List[String] = new util.ArrayList[String]()
          var boolAlert: Boolean = true
          breakable {
            logs.foreach(log => {
                events.add(log.evid)
              if ("clickItem".equals(log.evid)) {
                boolAlert = false
                break
              }

              if ("coupon".equals(log.evid)) {
                uids.add(log.uid)
                itemIds.add(log.itemid)
              }
            })
          }
          if (boolAlert == false || uids.size() < 3) {
            null
          }
          else {
            CouponAlertInfo(mid, uids, itemIds, events, System.currentTimeMillis())
          }
        }
      }

      val ds4: DStream[CouponAlertInfo] = ds3.filter(_ != null)

      ds4.foreachRDD(rdd => {
        rdd.foreachPartition(iter => {
          var client: JestClient = MyEsUtil.getClient
          val indexName = GmallConstants.ES_ALERT_INDEX_PREFIX + "_" + LocalDate.now()
          val formatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm")
          val doc: List[(String, CouponAlertInfo)] = iter.map(alter => {
            val time: LocalDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(alter.ts), ZoneId.of("Asia/Shanghai"))
            (alter.mid + "_" + time, alter)
          }).toList
          MyEsUtil.insertBulk(indexName, doc)
          MyEsUtil.close(client)
        })
      })
    }
  }
}